package com.zuoshengli.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zuoshengli.api.KafkaService;
import com.zuoshengli.api.OrderService;
import com.zuoshengli.api.StockService;
import com.zuoshengli.constant.RedisKeysContant;
import com.zuoshengli.dao.OrderMapper;
import com.zuoshengli.pojo.Order;
import com.zuoshengli.pojo.Stock;
import org.apache.dubbo.config.annotation.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;

/**
 * @Description TODO
 * @Author zuoshengli
 **/

@Service(version = "${demo.service.version}")
public class OrderServiceImpl implements OrderService {
    private Logger logger = LoggerFactory.getLogger(OrderServiceImpl.class);

    @Autowired
    private StockService stockService;

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private ObjectMapper objectMapper;

    @Value("${kafka.topic.topic-name}")
    private String kafkaTopic;


    @Override
    public int createWrongOrder(int sid) throws Exception {
        //检验库存
        Stock stock = checkStock(sid);
        //扣库存
        saleStock(stock);
        //创建订单
        int id = createOrder(stock);

        return 0;
    }

    @Override
    public int createOptimisticOrder(int sid) throws Exception {
        //检查库存
        Stock stock = checkStock(sid);
        //扣库存
        saleStockOptimistic(stock);
        //生成订单
        int id = createOrder(stock);
        return id;
    }

    @Override
    public int createOptimisticOrderUseRedis(int sid) {
        //检查库存
        Stock stock = checkStockByRedis(sid);
        //更新库存
        saleStockOptimisticByRedis(stock);
        //创建订单
        int id = createOrder(stock);
        return id;
    }

    @Override
    public void createOptimisticOrderUseRedisAndKafka(int sid) throws JsonProcessingException {
        //检查库存
        Stock stock = checkStockByRedis(sid);
        //利用kafka创建订单
        kafkaService.sendMessage(kafkaTopic, String.valueOf(sid));
        logger.info("send kafka success");

    }

    private Stock checkStockByRedis(int sid) {
        Integer count = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysContant.STOCK_COUNT+sid));
        Integer sale = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysContant.STOCK_SALE+sid));
        if (count.equals(sale)) {
            throw new RuntimeException("库存不足Redis currentCount=" + sale);
        }
        Integer version = Integer.parseInt(redisTemplate.opsForValue().get(RedisKeysContant.STOCK_VERSION+sid));
        Stock stock = new Stock();
        stock.setId(sid);
        stock.setCount(count);
        stock.setSale(sale);
        stock.setVersion(version);
        return stock;
    }

    private Stock checkStock(int sid) {
        Stock stock =stockService.getStockById(sid);
        if (stock.getSale().equals(stock.getCount())) {
            throw new RuntimeException("库存不足");
        }
        return stock;
    }

    private void saleStock(Stock stock) {
        stock.setSale(stock.getSale() + 1);
        stockService.updateStockById(stock);
    }

    private void saleStockOptimistic(Stock stock) {
        int count = stockService.updateStockByOptimistic(stock);
        if (count == 0) {
            throw new RuntimeException("库存不足");
        }
    }

    private void saleStockOptimisticByRedis(Stock stock) {
        int count = stockService.updateStockByOptimistic(stock);
        if (count == 0) {
            throw new RuntimeException("库存不足");
        }
        redisTemplate.opsForValue().increment(RedisKeysContant.STOCK_SALE+stock.getId(), 1);
        redisTemplate.opsForValue().increment(RedisKeysContant.STOCK_VERSION+stock.getId(), 1);
    }

    private int createOrder(Stock stock) {
        Order order = new Order();
        order.setSid(stock.getId());
        order.setName(stock.getName());
        return orderMapper.insertSelective(order);
    }

}
